package com.mengxin.secondkill.msgqueue;


import java.util.Map;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.mengxin.secondkill.constants.StreamConstans;
import com.mengxin.secondkill.message.req.AddOrderReq;
import com.mengxin.secondkill.service.OrderService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.stream.MapRecord;
import org.springframework.data.redis.connection.stream.Record;
import org.springframework.data.redis.connection.stream.RecordId;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.stereotype.Component;

/**
 * @author mengxin
 * @date 2020/12/24 14:06
 */
@Component
@Slf4j
public class StreamMessageListener implements StreamListener<String, MapRecord<String, String, Object>> {

    @Autowired
    OrderService orderService;

    @Autowired
    RedisTemplate<String,Object> redisTemplate;

    @Override
    public void onMessage(MapRecord<String, String, Object> message) {
        // 消息ID
        RecordId messageId = message.getId();
        // 消息的key和value
        Map<String, Object> body = message.getValue();
        log.info("Stream message;messageId={}, stream={}, body={}", messageId, message.getStream(), body);
        String key = message.getStream();
        if(StreamConstans.GOODS_BUY_STREAM_KEY.equals(key)){
            goodsBuy(body,message);
        }
    }

    private void goodsBuy(Map<String, Object> value, Record<String, ?> record) {
        Object dataJsonString = value.get("data");
        if(dataJsonString==null){
            log.error("商品购买请求生成订单，请求参数为空"+value);
            return;
        }
        JSONObject data = JSONObject.parseObject(dataJsonString.toString());
        AddOrderReq req = new AddOrderReq(data);
        orderService.addOrder(req,StreamConstans.GOODS_BUY_STREAM_GROUP, record);
    }

    /**
     * 通过RedisTemplate手动确认消息
     * @param group
     * @param record
     */
    public void ackMessage(String group,Record<String, ?> record){
        this.redisTemplate.opsForStream().acknowledge(group, record);
    }
}
